package org.fujene.communicate.searcher;

import java.util.TreeMap;
import java.util.Map.Entry;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;

import org.fujene.communicate.hash.VirtualContentHash;
import org.fujene.communicate.node.Node;
import org.fujene.communicate.repli.ReplicateCommunicator;
import org.fujene.logging.LogParser;
import org.fujene.memman.ReplicateMemoryManager;
import org.fujene.structure.ExtendTerm;
import org.fujene.structure.SearcherQueue;

public class SearcherHost implements Runnable{
	private static CopyOnWriteArrayList<Node> nodes;
	private static ReplicateMemoryManager memman;
	private static Node myNode;
	private SearcherQueue structure;
	private String term;
	private int count;
	private int contentType;
	private static int replication;
	private static LinkedBlockingQueue<SearcherQueue> queue;
	private int retryCount = 2;

	static{
		queue = new LinkedBlockingQueue<SearcherQueue>();
	}

	@Deprecated
	public SearcherHost(){}

	public SearcherHost(SearcherQueue structure, int count, int contentType){
		this.term = structure.getTerm(count).str();
		this.structure = structure;
		this.contentType = contentType;
		this.count = count;
	}

	public static void property(int repl, ReplicateMemoryManager manager, Node ownNode,
			CopyOnWriteArrayList<Node> list){
		replication = repl;
		memman = manager;
		myNode = ownNode;
		nodes = list;
	}

	public static void addToQueue(SearcherQueue newSearch){
		queue.add(newSearch);
	}

	public static int residue(){
		return queue.size();
	}

	public void run(){
		try{
			long hash = VirtualContentHash.getTermHash(term);
			TreeMap<Long, Node> hashList = new TreeMap<Long, Node>();
			Entry<Long, Node> currentEntry;
			Node termNode = null;
			for(Node eachNode: nodes)
				hashList.put(eachNode.getAttributeHash(hash), eachNode);
			currentEntry = hashList.floorEntry(Long.valueOf(hash));
			if(currentEntry == null) currentEntry = hashList.firstEntry();

			// search boosting
			for(int i = 0; i < replication; i++){
				if(currentEntry.getValue() == myNode){
					termNode = currentEntry.getValue();
					break;
				}
				termNode = currentEntry.getValue();
				currentEntry = hashList.lowerEntry(currentEntry.getKey());
				if(currentEntry == null) currentEntry = hashList.lastEntry();
			}

			LogParser.fine("Search - Node: " + termNode);
			if(termNode == myNode) structure.setResult(count, memman.searchTerm(term, contentType));
			else{
				byte[] extb = null;
				for(int i = 0; i < retryCount; i++){
					try{
						long time = LogParser.initTime();
						extb = ReplicateCommunicator.searchRemote(term, contentType, termNode);
						LogParser.timerFinest("Search / Post to remote", time);
						if(extb != null) structure.setResult(count, new ExtendTerm(extb));
						break;
					}catch(Exception e){
						LogParser.warning("Retrying search on next node.");
						continue;
					}
				}
			}
			// all searches will demolish if result's NULL
			if(structure.getResult(count) == null) structure.notFound();
			else{
				for(int i = 0; i < structure.getResult(count).invIndex.length; i++){
					if(structure.getResult(count).invIndex[i] == null){
						structure.notFound();
						break;
					}
				}
			}
		}finally{
			structure.countdown();
		}
	}
}
